package com.apps.sdses.flink141.demo;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableDescriptor;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.time.LocalDateTime;

/**
 * @author jiwei
 * @description
 * @date 2023/8/18 9:46
 */
public class TimeFrequencyAlignmentDemo {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        Table table = tableEnv.from(TableDescriptor
                .forConnector("filesystem")
                        .option("path", "src/main/resources/data/sensor.csv")
                        .option("format", "csv")
                        .schema(Schema.newBuilder()
                                .column("id", DataTypes.INT())
                                .column("sid", DataTypes.STRING())
                                .column("press", DataTypes.INT())
                                .column("ts", DataTypes.TIMESTAMP(3))
                                .watermark("ts", "ts - INTERVAL '1' SECOND")
                                .build()
                        )
                .build()
        );
        DataStream<Row> sourceDS = tableEnv.toDataStream(table);


        sourceDS
                .map(new MapFunction<Row, Tuple4<Integer, String, Integer, LocalDateTime>>() {
                    @Override
                    public Tuple4<Integer, String, Integer, LocalDateTime> map(Row value) throws Exception {
                        Integer id = value.<Integer>getFieldAs("id");
                        String sid = value.<String>getFieldAs("sid");
                        Integer pressure = value.<Integer>getFieldAs("press");
                        LocalDateTime ts = value.<LocalDateTime>getFieldAs("ts");
                        return Tuple4.of(id, sid, pressure, ts);
                    }
                })
                .keyBy(new KeySelector<Tuple4<Integer, String, Integer, LocalDateTime>, String>() {
                    @Override
                    public String getKey(Tuple4<Integer, String, Integer, LocalDateTime> value) throws Exception {
                        return value.f1;
                    }
                })
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .sum(2)
                .print();

        env.execute();
    }
}
